these are unrelated changes. When I looked at the generated codes, I found Literal generated verbose codes, so I simplified it a little bit.

Is it better to split into two PRs? To put two changes into one PR may not be easy to understand diffs of performance results.

ok I reverted. The result is almost the same so I didn't update

also test the grouping performance, not only aggregating.

Is there any reason to add grouping operation?

aggregate without grouping is not a common use case

i see

Actually, I think we should also have a test case for aggregation without group by.

for "back-to-back map", the logic is so simple that the code generated by Dataset is less efficient than RDD. RDD just adds 1 to the input Long, the only overhead is boxing, while Dataset generates code like this:

boolean mapelements_isNull = true;
long mapelements_value = -1L;
if (!false) {
  mapelements_argValue = range_value;
  mapelements_isNull = false;
  if (!mapelements_isNull) {
    Object mapelements_funcResult = null;
    mapelements_funcResult = mapelements_obj.apply(mapelements_argValue);
    if (mapelements_funcResult == null) {
      mapelements_isNull = true;
    } else {
      mapelements_value = (Long) mapelements_funcResult;
    }
  }
}

Dataset still has the boxing overhead, but its code is more verbose. And Dataset has to write the long to un unsafe row at last, which is another overhead. These are the reasons why Dataset is slower than RDD for this simple case.

IIUC, an signature of apply() is Object apply(Object). It also introduces additional boxing overhead from long to Long.
To reduce these boxing and unboxing overhead, we need to use more concrete signature (e.g. long apply(long) to call a lambda function.

the method signature in Dataset is: def map[U : Encoder](f: T => U), unless we create primitive version methods, e.g. def map(f: T => Long), I can't think of an easy way to get the concrete signature.

BTW, I think the best solution is to analyze the byte code(class file) of the lambda function, and turn it into expressions.

I noticed that Scala compiler automatically generates primitive version. Current Spark eventually calls primitive version thru generic version Object apply(Object).

Here is a simple example. When we compile the following Dataset program, we can find that the following class is generated by scalac. Scalac automatically generates a primitive version int apply$mcII$sp(int) that can be called by int apply(int).
We could infer this signature in Catalyst for simple cases.

Of course, I totally agree that the best solution is to analyze byte code and turn it into expression. This was already prototyped. Do you think it is good time to make this prototype more robust now?

test("ds") {
  val ds = sparkContext.parallelize((1 to 10), 1).toDS
  ds.map(i => i * 7).show
}

$ javap -c Test\$\$anonfun\$5\$\$anonfun\$apply\$mcV\$sp\$1.class
Compiled from "Test.scala"
public final class org.apache.spark.sql.Test$$anonfun$5$$anonfun$apply$mcV$sp$1 extends scala.runtime.AbstractFunction1$mcII$sp implements scala.Serializable {
  public static final long serialVersionUID;

  public final int apply(int);
    Code:
       0: aload_0
       1: iload_1
       2: invokevirtual #18                 // Method apply$mcII$sp:(I)I
       5: ireturn

  public int apply$mcII$sp(int);
    Code:
       0: iload_1
       1: bipush        7
       3: imul
       4: ireturn

  public final java.lang.Object apply(java.lang.Object);
    Code:
       0: aload_0
       1: aload_1
       2: invokestatic  #29                 // Method scala/runtime/BoxesRunTime.unboxToInt:(Ljava/lang/Object;)I
       5: invokevirtual #31                 // Method apply:(I)I
       8: invokestatic  #35                 // Method scala/runtime/BoxesRunTime.boxToInteger:(I)Ljava/lang/Integer;
      11: areturn

  public org.apache.spark.sql.Test$$anonfun$5$$anonfun$apply$mcV$sp$1(org.apache.spark.sql.Test$$anonfun$5);
    Code:
       0: aload_0
       1: invokespecial #42                 // Method scala/runtime/AbstractFunction1$mcII$sp."<init>":()V
       4: return
}

ah, scala compiler is smart! I think we can create a ticket to optimize this, i.e. call the primitive apply version, and update the benchmark result.

For byte code analysis, let's discuss about it in the ticket later.

Sure, I will create a JIRA ticket for this optimization.

For byte code analysis, let's restart discuss about it the JIRA entry.

For "back-to-back filter", Dataset will deserialize the input row to an object and apply the condition function. When the deserialization becomes no-op, Dataset runs almost the same RDD code like the RDD case. So in this case, RDD and Dataset has similar performance.

For "aggregate", Dataset use AppendColumnsExec to generate the grouping key, which will do an extra copy(the unsafe row joiner). This makes Dataset slower than RDD.

It would be good to update aggregate.